package zy.learn.demo.structuredstreaming.source

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.{OutputMode, Trigger}

object KafkaSource2 {
  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().set("spark.sql.shuffle.partitions", "3")

    val spark = SparkSession.builder()
      .master("local[2]")
      .config(sparkConf)
      .appName("KafkaSource聚合")
      .getOrCreate()

    import spark.implicits._

    /*
     kafka-topics.sh --create --bootstrap-server co7-203:9092 --topic topic1 --replication-factor 1 --partitions 1
     kafka-console-producer.sh --broker-list co7-203:9092 --sync --topic topic1
     输入数据：
          Hanks 35 male;Mina 26 female;Tracy 33 female;Jimi 30 male
          Tony 30 male;BlackWidow 28 female;Hulk 40 male;Natasha 34 female
          Tina 20 female
     结果输出：
+------+---------+-------+-------+
|sex   |count_age|sum_age|avg_age|
+------+---------+-------+-------+
|female|2        |59     |29.5   |
|male  |2        |65     |32.5   |
+------+---------+-------+-------+
|female|4        |121    |30.25  |
|male  |4        |135    |33.75  |
+------+---------+-------+-------+
|female|5        |141    |28.2   |
+------+---------+-------+-------+
     */

    // 得到的 df 的 schema 是固定的: key,value,topic,partition,offset,timestamp,timestampType
    val df = spark.readStream
      .format("kafka") // 设置 kafka 数据源
      .option("kafka.bootstrap.servers", "co7-203:9092,co7-204:9092,co7-205:9092")
      .option("subscribe", "topic1") // 也可以订阅多个主题:   "topic1,topic2"
      .load
      .selectExpr("cast(value as string)") // 选取字段，且将 value由byte转化为string
      .as[String].flatMap(_.split(";"))
      .map(value => {
        val fields = value.split(" ")
        (fields(0), fields(1).toInt, fields(2))
      })

    df.toDF("name", "age", "sex").createOrReplaceTempView("table1")

    val resultDF = spark.sql(
      """
        | select sex, count(age) as count_age, sum(age) as sum_age, avg(age) as avg_age
        |   from table1
        |  group by sex
        |""".stripMargin)

    resultDF.writeStream
      .outputMode(OutputMode.Update()) // 只输出变化的内容
//      .outputMode(OutputMode.Complete()) // 输出全部的内容，不论是否变化，后续测试watermark对complete mode的影响
      .option("truncate", false)  // 将打印的内容显示完全
      .format("console")
//      .trigger(Trigger.Continuous(1000))    // Continuous processing does not support Aggregate operations.
      .start
      .awaitTermination()
  }
}
